package com.yanggu.bigdata.realtime.app.dwm

import cn.hutool.core.date.{DateTime, DateUtil}
import cn.hutool.json.JSONUtil
import com.alibaba.fastjson.JSONObject
import com.yanggu.bigdata.realtime.app.function.DimJoinRichAsyncFunction
import com.yanggu.bigdata.realtime.bean.{OrderDetail, OrderInfo, OrderWide}
import com.yanggu.bigdata.realtime.common.GmallConfig.KAFKA_BROKER_LIST
import com.yanggu.bigdata.realtime.utils.KafkaUtil
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

import java.util.concurrent.TimeUnit

/**
 * DWM层订单明细宽表
 * 使用订单表JOIN订单明细表, 同时使用维度表来补充明细表的数据
 * 外部数据源的查询常常是流式计算的性能瓶颈
 * 因此常见的做法有旁路缓存和异步IO
 */
/**
 * 数据流向 client -> Springboot -> MySQL -> FlinkApp -> Kafka(ods_base_db) -> FlinkApp -> Kafka(dwd_事实表)、Hbase(DIM) -> FlinkApp -> Kafka(dwm_事实宽表)
 * 程 序 MockDB -> MySQL -> flink-cdc -> Kafka(ods_base_db) -> Flink(BaseDBApp) -> Kafka(dwd_事实表)、Hbase(DIM) -> Flink(OrderWideApp) -> Kafka(dwm_事实宽表)
 *                                       Kafka、zk                                 Kafka、zk、hbase、hdfs、phoenix      redis、phoenix       Kafka、zk
 */
object OrderWideApp {

  def main(args: Array[String]): Unit = {
    //1. 设置执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //2.读取 Kafka 订单和订单明细主题数据
    //2.1 订单表 读取topic:dwd_order_info中的数据, 提取Watermark
    val keyedOrderInfoDataStream = KafkaUtil.getKafkaDataStream(env, KAFKA_BROKER_LIST, "dwd_order_info", "OrderWideApp")
      .map(data => {
        val orderInfo = JSONUtil.toBean(data, classOf[OrderInfo])
        val createTime = DateUtil.parseDateTime(orderInfo.getCreate_time)
        //设置创建日期
        orderInfo.setCreate_date(DateUtil.formatDate(createTime))
        //设置小时
        orderInfo.setCreate_hour(DateUtil.formatTime(createTime))
        //设置创建时间戳
        orderInfo.setCreate_ts(createTime.getTime)
        orderInfo
      })
      //提取Watermark
      .assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps[OrderInfo]
        .withTimestampAssigner(new SerializableTimestampAssigner[OrderInfo] {
          override def extractTimestamp(element: OrderInfo, recordTimestamp: Long): Long = element.getCreate_ts
        })
      ).setParallelism(1)
      //按照orderId进行分区
      .keyBy(_.getId)

    //2.2 订单明细 读取topic:dwd_order_detail中的数据, 提取Watermark
    val keyedOrderDetailDataStream = KafkaUtil.getKafkaDataStream(env, KAFKA_BROKER_LIST, "dwd_order_detail", "OrderWideApp")
      .map(data => {
        val orderDetail = JSONUtil.toBean(data, classOf[OrderDetail])
        val createTime = DateUtil.parseDateTime(orderDetail.getCreate_time)
        //设置创建日期
        orderDetail.setCreate_ts(createTime.getTime)
        orderDetail
      })
      //提取Watermark
      .assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps[OrderDetail]
        .withTimestampAssigner(new SerializableTimestampAssigner[OrderDetail] {
          override def extractTimestamp(element: OrderDetail, recordTimestamp: Long): Long = element.getCreate_ts
        })
      ).setParallelism(1)
      //按照orderId进行分区
      .keyBy(_.getOrder_id)

    //订单流去intervalJoin订单明细流
    val orderWideDataStream = keyedOrderInfoDataStream
      .intervalJoin(keyedOrderDetailDataStream)
      .between(Time.seconds(-10L), Time.seconds(5L))
      .process((left: OrderInfo, right: OrderDetail, _: ProcessJoinFunction[OrderInfo, OrderDetail, OrderWide]#Context, out: Collector[OrderWide]) => {
        out.collect(new OrderWide(left, right))
      })

    //控制台打印输出订单宽表数据
    //orderWideDataStream.print("OrderWideData>>>>>").setParallelism(1)

    //3. 补充订单宽表事实表的维度数据
    //3.1 补充DIM_USER_INFO维度表数据
    val orderWideWithDimUserInfo = AsyncDataStream.unorderedWait(orderWideDataStream, new DimJoinRichAsyncFunction[OrderWide] {

      override def tableName: String = "DIM_USER_INFO"

      override def getKey(input: OrderWide): String = input.getUser_id.toString

      //使用用户维度表数据染色订单宽表数据
      override def join(input: OrderWide, dimInfo: JSONObject): Unit = {
        //phoenix中字段都是大写
        //1. age
        val age = DateTime.now().year() - DateUtil.parseDateTime(dimInfo.getString("BIRTHDAY")).year()
        input.setUser_age(age)
        //2. gender
        input.setUser_gender(dimInfo.getString("GENDER"))
      }
    }, 60, TimeUnit.SECONDS, 100)
    //orderWideWithDimUserInfo.print("orderWideWithDimUserInfo>>>>>").setParallelism(1)

    //3.2 补充地区维度数据
    val orderWideWithDimProvinceInfo = AsyncDataStream.unorderedWait(orderWideWithDimUserInfo, new DimJoinRichAsyncFunction[OrderWide] {

      /**
       * 抽象字段, 返回Hbase中的维度表名
       *
       * @return
       */
      override def tableName = "DIM_BASE_PROVINCE"

      /**
       * 通过输入数据获取维度表的主键
       *
       * @param input 事实宽表数据
       * @return
       */
      override def getKey(input: OrderWide) = input.getProvince_id.toString

      /**
       * 根据查询得到的维度表数据来补充事实表的字段
       * 专业术语叫使用维度表染色事实表
       *
       * @param input   事实宽表数据
       * @param dimInfo 维度表数据
       */
      override def join(input: OrderWide, dimInfo: JSONObject): Unit = {
        //name、area_code、iso_code、iso_3166_2
        input.setProvince_name(dimInfo.getString("NAME"))
        input.setProvince_area_code(dimInfo.getString("AREA_CODE"))
        input.setProvince_iso_code(dimInfo.getString("ISO_CODE"))
        input.setProvince_3166_2_code(dimInfo.getString("ISO_3166_2"))
      }
    }, 60, TimeUnit.SECONDS, 100)
    //orderWideWithDimProvinceInfo.print("orderWideWithDimProvinceInfo>>>>>>>").setParallelism(1)

    //3.3 补充SKU数据
    val orderWideWithDimSKUInfo = AsyncDataStream.unorderedWait(orderWideWithDimProvinceInfo, new DimJoinRichAsyncFunction[OrderWide] {

      /**
       * 抽象字段, 返回Hbase中的维度表名
       *
       * @return
       */
      override def tableName: String = "DIM_SKU_INFO"

      /**
       * 通过输入数据获取维度表的主键
       *
       * @param input 事实宽表数据
       * @return
       */
      override def getKey(input: OrderWide): String = input.getSku_id.toString

      /**
       * 根据查询得到的维度表数据来补充事实表的字段
       * 专业术语叫使用维度表染色事实表
       *
       * @param input   事实宽表数据
       * @param dimInfo 维度表数据
       */
      override def join(input: OrderWide, dimInfo: JSONObject): Unit = {
        input.setSku_name(dimInfo.getString("SKU_NAME"))
        input.setSpu_id(dimInfo.getLong("SPU_ID"))
        input.setCategory3_id(dimInfo.getLong("CATEGORY3_ID"))
        input.setTm_id(dimInfo.getLong("TM_ID"))
      }
    }, 60L, TimeUnit.SECONDS, 100)
    //orderWideWithDimSKUInfo.print("orderWideWithDimSKUInfo>>>>>").setParallelism(1)

    //3.4 补充SPU维度数据
    val orderWideWithDimSPUInfo = AsyncDataStream.unorderedWait(orderWideWithDimSKUInfo, new DimJoinRichAsyncFunction[OrderWide] {
      /**
       * 抽象字段, 返回Hbase中的维度表名
       *
       * @return
       */
      override def tableName: String = "DIM_SPU_INFO"

      /**
       * 通过输入数据获取维度表的主键
       *
       * @param input 事实宽表数据
       * @return
       */
      override def getKey(input: OrderWide): String = input.getSpu_id.toString

      /**
       * 根据查询得到的维度表数据来补充事实表的字段
       * 专业术语叫使用维度表染色事实表
       *
       * @param input   事实宽表数据
       * @param dimInfo 维度表数据
       */
      override def join(input: OrderWide, dimInfo: JSONObject): Unit = {
        input.setSpu_name(dimInfo.getString("SPU_NAME"))
      }
    }, 60L, TimeUnit.SECONDS, 100)
    //orderWideWithDimSPUInfo.print("orderWideWithDimSPUInfo>>>>>>").setParallelism(1)

    //3.5 补充品牌维度数据
    val orderWideWithDimTMInfo = AsyncDataStream.unorderedWait(orderWideWithDimSPUInfo, new DimJoinRichAsyncFunction[OrderWide] {
      /**
       * 抽象字段, 返回Hbase中的维度表名
       *
       * @return
       */
      override def tableName: String = "DIM_BASE_TRADEMARK"

      /**
       * 通过输入数据获取维度表的主键
       *
       * @param input 事实宽表数据
       * @return
       */
      override def getKey(input: OrderWide): String = input.getTm_id.toString

      /**
       * 根据查询得到的维度表数据来补充事实表的字段
       * 专业术语叫使用维度表染色事实表
       *
       * @param input   事实宽表数据
       * @param dimInfo 维度表数据
       */
      override def join(input: OrderWide, dimInfo: JSONObject): Unit = {
        //tm_name设置
        input.setTm_name(dimInfo.getString("TM_NAME"))
      }
    }, 60L, TimeUnit.SECONDS, 100)

    //3.6 补充3级分类维度数据
    val orderWideWithDimCategory3Info = AsyncDataStream.unorderedWait(orderWideWithDimTMInfo, new DimJoinRichAsyncFunction[OrderWide] {
      /**
       * 抽象字段, 返回Hbase中的维度表名
       *
       * @return
       */
      override def tableName: String = "DIM_BASE_CATEGORY3"

      /**
       * 通过输入数据获取维度表的主键
       *
       * @param input 事实宽表数据
       * @return
       */
      override def getKey(input: OrderWide): String = input.getCategory3_id.toString

      /**
       * 根据查询得到的维度表数据来补充事实表的字段
       * 专业术语叫使用维度表染色事实表
       *
       * @param input   事实宽表数据
       * @param dimInfo 维度表数据
       */
      override def join(input: OrderWide, dimInfo: JSONObject): Unit = {
        input.setCategory3_name(dimInfo.getString("NAME"))
      }
    }, 60L, TimeUnit.SECONDS, 1000)

    //4. 输出
    //把数据输出到Kafka的dwm层, 订单宽表
    orderWideWithDimCategory3Info.print("orderWideWithDimCategory3Info>>>>>>").setParallelism(1)
    orderWideWithDimCategory3Info
      .map(JSONUtil.toJsonStr(_))
      .addSink(KafkaUtil.getKafkaSink(KAFKA_BROKER_LIST, "dwm_order_wide"))

    //5. 启动Job
    env.execute("OrderWideApp Job")

  }

}
